package cn.ac.iscas.service.transformer;

import cn.ac.iscas.base.utils.SpringUtil;
import cn.ac.iscas.domain.flink.FlinkColumnType;
import cn.ac.iscas.domain.source.FlinkSourceColumn;
import cn.ac.iscas.service.queue.flink.FlinkTableQueueService;

import java.util.List;
import java.util.Map;
import java.util.UUID;
import java.util.concurrent.ConcurrentHashMap;


/**
 * @description:
 * @param:
 * "parameters": {
 *         "null_columns": [
 *             {
 *                 "column": "column1",
 *                 "fill_value": "empty",
 *                 "type": "STRING",
 *                 "label": "column1_fill"
 *             },
 *             {
 *                 "column": "column2",
 *                 "fill_value": 0,
 *                 "type": "INTEGER",
 *                 "label": "column2_fill"
 *             }
 *         ]
 * }
 * @return:
 * @author LJian
 * @date: 2022/11/16 14:50
 */
public class FullfilNullTransformer {
    private final static String ID = "fullfil_null";
    private final static String COLUMN_NAME_KEY = "column";
    private final static String LABEL_NAME_KEY = "label";
    private final static String NULL_COLUMNS_NAME_KEY = "null_columns";
    private final static String COLUMN_TYPE_KEY = "type";
    private final static String FILL_VALUE_KEY = "fill_value";
    private String jobId;
    private String componentId;
    private String frontTableName;
    private List<FlinkSourceColumn> frontColumns;

    private String currentTableName;
    private List<FlinkSourceColumn> currentColumns;

    public FullfilNullTransformer(String jobId, String taskId) {
        this.jobId = jobId;
        this.componentId = taskId;
    }

    public FullfilNullTransformer(String jobId, String taskId, String frontTableName, List<FlinkSourceColumn> frontColumns) {
        this.jobId = jobId;
        this.componentId = taskId;
        this.frontTableName = frontTableName;
        this.frontColumns = frontColumns;
        this.currentColumns = frontColumns;
    }


    public FullfilNullTransformer() {

    }

    //CREATE VIEW c11ddebe.c11ddebe.fullfil3 AS ( SELECT *, IFNULL(score, 80) AS name FROM c11ddebe.c11ddebe.class_table );
    public String getSql(Map<String, Object> parameters) {
        this.currentTableName = String.format("%s.%s.%s_%s", jobId, jobId, ID, UUID.randomUUID().toString().substring(0,6));
        String createViewSql = String.format("CREATE VIEW %s AS ", this.currentTableName);
        //List<Map<String, Object>> nullFilParameters = ParameterUtils.getRequiredList(parameters, NULL_COLUMNS_NAME_KEY);
        //Map<String, Object> nullFilParameter = ParameterUtils.getRequiredMap(parameters, NULL_COLUMNS_NAME_KEY);
        Map<String, Object> nullFilParameter = parameters;
        StringBuffer nullFilSqlSb = new StringBuffer();
        //for (Map<String, Object> nullFilParameter : nullFilParameters) {
            String column = nullFilParameter.getOrDefault(COLUMN_NAME_KEY, "").toString();
            String columnType = nullFilParameter.getOrDefault(COLUMN_TYPE_KEY, "").toString();
            String fillValue = nullFilParameter.getOrDefault(FILL_VALUE_KEY, "").toString();
            String label = nullFilParameter.getOrDefault(LABEL_NAME_KEY, "").toString();

            //if (Double.parseDouble(fillValue))
            if (columnType == null || columnType.isEmpty()) {
                //continue;
                String reg = "^-?[0-9]+(.[0-9]+)?$";
                if (!fillValue.matches(reg)) {
                    columnType = FlinkColumnType.STRING.getName();
                }
            }
            if (columnType.equalsIgnoreCase(FlinkColumnType.STRING.getName())) {
                nullFilSqlSb.append(String.format(" IFNULL(%s, '%s') AS %s , ", column, fillValue, label));
            } else {
                nullFilSqlSb.append(String.format(" IFNULL(%s, %s) AS %s , ", column, fillValue, label));
            }
            FlinkSourceColumn newColumn = new FlinkSourceColumn();
            newColumn.setType(columnType);
            newColumn.setName(label);
            newColumn.setPrimaryKey(false);
            this.currentColumns.add(newColumn);
       // }

        String nullFilSql = nullFilSqlSb.toString();
        nullFilSql = nullFilSql.substring(0, nullFilSql.lastIndexOf(","));
        String selectSql = String.format(" ( SELECT *, %s FROM %s ) ", nullFilSql, frontTableName);
        createViewSql += selectSql;


        updateSource();
        return createViewSql;
    }

    public void updateSource() {
        FlinkTableQueueService flinkTableQueueService = SpringUtil.getBean(FlinkTableQueueService.class);
        Map<String, List<FlinkSourceColumn>> tableColumns = new ConcurrentHashMap<>(8);
        tableColumns.put(this.currentTableName, this.currentColumns);
        flinkTableQueueService.put(this.jobId, this.componentId, tableColumns);
    }
}
